package rxgo

import (
	"context"
	"fmt"
	"reflect"
	"time"
)

type filterOperator struct {
	operatorPtr func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool)
}

func (ftop filterOperator) op(ctx context.Context, obs *Observable) {
	in := obs.pred.outflow
	out := obs.outflow
	go func() {
		end := false
		for x := range in {
			if end {
				break
			}
			value_x := reflect.ValueOf(x)
			if e, ok := x.(error); ok && !obs.flip_accept_error {
				obs.sendToFlow(ctx, e, out)
				continue
			}
			if ftop.operatorPtr(ctx, obs, value_x, out) {
				end = true
			}
		}
		if obs.flip != nil {
			buffer := (reflect.ValueOf(obs.flip))
			if buffer.Kind() != reflect.Slice {
				panic("flip is not buffer")
			}
			for i := 0; i < buffer.Len(); i++ {
				obs.sendToFlow(ctx, buffer.Index(i).Interface(), out)
			}
		}
		obs.closeFlow(out)
	}()
}

var count = 0

//Debounce :实现数据的延迟发送
func (parent *Observable) Debounce(timespan time.Duration) (obs *Observable) {
	obs = parent.newFilterObservable("debounce")
	obs.time = timespan
	obs.operator = debounceOperator
	return obs
}

var debounceOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		go func() {
			tempCount := count
			starttime := time.Now()
			for {
				aftertime := time.Since(starttime)
				if aftertime >= obs.time {
					break
				}
			}
			select {
			case <-ctx.Done():
				return
			default:
				if tempCount == count {
					obs.sendToFlow(ctx, val.Interface(), out)
				}
			}
		}()
		return false
	},
}

//Distinct ：实现数据去重
func (parent *Observable) Distinct() (obs *Observable) {
	obs = parent.newFilterObservable("distinct")
	obs.dismap = make(map[string]bool)
	obs.operator = distinctOperator
	return obs
}

var distinctOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		itemstring := fmt.Sprintf("%v", val)

		if _, ok := obs.dismap[itemstring]; !ok {
			obs.dismap[itemstring] = true
			obs.sendToFlow(ctx, val.Interface(), out)
		}
		return false
	},
}

//ElementAt ： 去除对应位置的数据
func (parent *Observable) ElementAt(num int) (obs *Observable) {
	obs = parent.newFilterObservable("elementAt")
	count = 0
	obs.elementatnum = num
	obs.operator = elementAtOperator
	return obs
}

var elementAtOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		if count == obs.elementatnum {
			obs.sendToFlow(ctx, val.Interface(), out)
			return true
		}
		count++
		return false
	},
}

// First ： 输出第一个数据
func (parent *Observable) First() (obs *Observable) {
	obs = parent.newFilterObservable("first")
	obs.operator = firstOperator
	return obs
}

var firstOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		obs.sendToFlow(ctx, val.Interface(), out)
		return true
	},
}

//IgnoreElements ： 忽略全部数据
func (parent *Observable) IgnoreElements() (obs *Observable) {
	obs = parent.newFilterObservable("ignoreElements")
	obs.operator = ignoreOperator
	return obs
}

var ignoreOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		return false
	},
}

//Last ： 输出最后一个数据
func (parent *Observable) Last() (obs *Observable) {
	obs = parent.newFilterObservable("last")
	obs.operator = lastOperator
	return obs
}

var lastOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		obs.flip = append([]interface{}{}, val.Interface())
		return false
	},
}

//Skip ：跳过前几个数据
func (parent *Observable) Skip(num int) (obs *Observable) {
	obs = parent.newFilterObservable("Skip")
	count = 0
	obs.skipnum = num
	obs.operator = skipOperator
	return obs
}

var skipOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		count++
		if count > obs.skipnum {
			obs.sendToFlow(ctx, val.Interface(), out)
		}
		return false
	},
}

//SkipLast ： 跳过最后几个数据
func (parent *Observable) SkipLast(num int) (obs *Observable) {
	obs = parent.newFilterObservable("SkipLast")
	count = 0
	obs.skiplastnum = num
	lasts = nil
	obs.operator = lastSkipOperator
	return obs
}

var lasts []interface{}

var lastSkipOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		if count == obs.skiplastnum {
			obs.sendToFlow(ctx, lasts[0], out)
			lasts = lasts[1:]
		} else {
			count++
		}
		lasts = append(lasts, val.Interface())
		return false
	},
}

//Take ：选取前几个元素输出
func (parent *Observable) Take(num int) (obs *Observable) {
	obs = parent.newFilterObservable("take.n")
	count = 0
	obs.takenum = num
	obs.operator = takeOperator
	return obs
}

var takeOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		count++
		if count > obs.takenum {
			return true
		}
		obs.sendToFlow(ctx, val.Interface(), out)
		return false
	},
}

//TakeLast ： 获取最后n个元素
func (parent *Observable) TakeLast(num int) (obs *Observable) {
	obs = parent.newFilterObservable("takeLast")
	count = 0
	lasts = nil
	obs.takelastnum = num
	obs.operator = lastTakeOperator
	return obs
}

var lastTakeOperator = filterOperator{
	operatorPtr: func(ctx context.Context, obs *Observable, val reflect.Value, out chan interface{}) (end bool) {
		count++
		if count <= obs.takelastnum {
			lasts = append(lasts, val.Interface())
		} else {
			lasts = lasts[1:]
			lasts = append(lasts, val.Interface())
		}
		obs.flip = lasts
		return false
	},
}

func (parent *Observable) newFilterObservable(name string) (obs *Observable) {
	obs = newObservable()
	obs.Name = name

	parent.next = obs
	obs.pred = parent
	obs.root = parent.root

	obs.buf_len = BufferLen
	return obs
}
